package com.aotain.jupiter.flink;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.IPSearcher;
import com.aotain.common.CommonFunction;

public class SipSoucreStreamStat implements MapFunction<String, Tuple2<String,String>> {

	String dbJson ;
	public SipSoucreStreamStat(String dbJson){
		this.dbJson = dbJson;

	}

	@Override
	public Tuple2<String, String> map(String value) throws Exception {
		// TODO Auto-generated method stub

		String refer = "";             String sourceAreaCountry = null;    String domainname = "";                    
		String dip = "";               String sourceAreaCity = null;       String url = "";                           
		String sip = "";               String sourceAreaName = null;       String type = "";                          
		String userAgent = "";         String destAreaCountry = null;      String accesstime = "";         
		String cookie = "";            String destAreaCity = null;         String idcno = "";              
		String dipL = "";              String destAreaName = null;         String post = "0";	          
		String sipL = "";              String gisstr = null;               String dsip = "";               
		String dport = "";             String gisstrBase64 = null;         String areaName_d = null;       
		String sport = "";             String Sourcelon = null;            long upStreamOctets = 0;        
		int status ;                   long upStreamPacket = 0;            String areaName_s = null;           
		String Sourcelat = null;       String sourceAreaId = null;         String destAreaId = null;          
		String Destlon = null;         String sourceAreaProvinceId = null; String destAreaProvinceId = null;  
		String Destlat = null;         String sourceAreaCityId = null;     String destAreaCityId = null;      
		String delaytime = null;
		Tuple2<String,String> ret = new Tuple2<String,String>();

		if(!"".equals(value)) {
			String[] splits = value.split("\\|",-1);
			if(splits.length>23) {
				String userAccount = splits[1];
				String ipRex = "([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}";   
				Pattern pattern = Pattern.compile(ipRex);   
				Matcher matcher = pattern.matcher(userAccount);  
				if(matcher.matches()) {
					if("1".equals(splits[22]) && "6".equals(splits[7]) && splits.length >43) {//http
						//if(splits.length ==43) {//http
						dipL = splits[9];
						dport = splits[10];
						sipL = splits[12];
						sport = splits[13];
						domainname = splits[32];
						url = domainname + "/" + splits[28];
						type = splits[7]; 
						accesstime= splits[14];
						idcno = "446605";
						upStreamOctets = Long.valueOf(splits[18]);
						upStreamPacket = Long.valueOf(splits[20]);
						status = Integer.valueOf(splits[23]);
						userAgent = splits[26];
						refer = splits[37];
						cookie = splits[42];
						if(splits.length>43) {
							post  = splits[43];
						}
					}else{//非Http
						dipL = splits[9];
						dport = splits[10];
						sipL = splits[12];
						sport = splits[13];
						domainname = "";
						url = "";
						type = splits[7]; 
						accesstime= splits[14];
						idcno = "446605";
						upStreamOctets = Long.valueOf(splits[18]);
						upStreamPacket = Long.valueOf(splits[20]);
						status = Integer.valueOf(splits[23]);
					}
					dip = CommonFunction.int2ip(Long.parseLong(dipL));
					sip = CommonFunction.int2ip(Long.parseLong(sipL));
					delaytime = 0 + "";
					long currenttime = System.currentTimeMillis();
					if(accesstime != null && !"".equals(accesstime)) {
						delaytime = currenttime/1000 - Long.parseLong(accesstime.trim()) + "";
					}

					if(!"".equals(dipL) && !"".equals(dport) && !"".equals(sipL) && !"".equals(sport) ) {

						IPSearcher ipsearcher = IPSearcher.getInstance(dbJson);
						IPDatabase SourceArea = ipsearcher.SearchIPDB(sip);
						IPDatabase DestArea = ipsearcher.SearchIPDB(dip);

						if(DestArea != null) {
							destAreaCountry = DestArea.getCountryName();
							destAreaCity = DestArea.getCityName();
							//						sourceAreaProvince = SourceArea.getProviceName();
							if(destAreaCity != null) {
								destAreaName  = destAreaCity;
								destAreaId = DestArea.getCityID()+"";
							} else {
								destAreaName = destAreaCountry;
								destAreaId = DestArea.getCountryID()+"";
							}
							destAreaCityId = sourceAreaId;
							Destlon = String.valueOf(DestArea.getLon());
							Destlat = String.valueOf(DestArea.getLat());

							if(!"".equals(DestArea.getProviceID()) && DestArea.getProviceID() !=0 ) {
								destAreaId = DestArea.getProviceID()+"";
								areaName_d = DestArea.getProviceName().replace("省", "");
								areaName_d = areaName_d.replace("市", "");
							} else {
								areaName_d = DestArea.getCountryName();
								destAreaId = DestArea.getCountryID()+"";
							}
							destAreaProvinceId = destAreaId;
						}   

						if(SourceArea != null) {    
							sourceAreaCountry = SourceArea.getCountryName();
							sourceAreaCity = SourceArea.getCityName();
							if(sourceAreaCity != null) {
								sourceAreaName = sourceAreaCity;
							} else {   
								sourceAreaName = sourceAreaCountry;
							}
							Sourcelon = String.valueOf(SourceArea.getLon());//经度
							Sourcelat = String.valueOf(SourceArea.getLat());//纬度
						}  

						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null) {
							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + destAreaCountry + "#" + destAreaId + "#" +areaName_d + "#" + destAreaCityId + "#" + destAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							ret.setFields(sip, dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" +domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" + post+ "|" +refer);
							//							collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno,upStreamOctets,upStreamPacket,status,userAgent,cookie,post,refer));
							//							collector.ack(tuple);
						}
					}
				}
			}

		}

		return ret;
	}

}
